﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using MessageServer.config;
using MessageServer.threadpool;
using MessageServer.utilities;
using MessageServer.pool;
using System.Collections;
using System.Net.Sockets;
using MessageServer.channels;
using MessageServer.channels.socket;
using MessageServer.logging;
using System.Threading;

namespace MessageServer.channels.socket
{
    /**
     * This class is the standard sink used for sinking events streamed down from
     * a serverchannel. This sink handles bound channel state event and starts up
     * workers for accepting connections from the channel and reading data from the
     * channel.
     * */
    class ServerChannelSink : Sink
    {
        TuningParams _config;               //The parameters used to tune the threads and workers
        ServerWorker _mainWorker;           //The main worker that listens for incoming connections
        DistributionPool _workers;          //The pool of workers to which work is distributed
        Executor _svrExecutor;              //The executor used to execute the listener
        //WorkerSelector _selector;           //The selector that check if work is present for a worker and assigns it
        DistributionPool _selectors;          //The selector will be created as a distribution pool to which workers are distributed
        Executor _clientExecutor;             //the executor that will be used by every selector to schedule work for socketworkers

        /**
         * The ServerChannelSink is setup for the tuning parameters configured. This does
         * not start up any of the executors. This just initializes them and waits
         * for an event to start up the executors. This sets up the worker pool based
         * on the Maximum Concurrent connections that are possible and the number of
         * socket workers that are configured.
         * */
        public ServerChannelSink(TuningParams config)
        {
            ChannelLogger.LOG.LogDebug("ServerChannelSink:ServerChannelSink:Constructing a sink for socket");
            _config = config;
            _svrExecutor = new AsynchronousExecutor();
            String nm = typeof(SocketWorker).Name;
            //All selectors will use the same threadpoolexecutor that will allow different
            //workers to schedule work into the same pool and hence restrict the total number
            //of threads that are running.
            _clientExecutor = new ThreadPoolExecutor(nm, config.WaitForThread, _config.MaxSocketWorkers,
                        config.MaxSocketWorkers);
            _selectors = new DistributionPool(new WorkerSelectorCreator(config, _clientExecutor), config.WaitForThread,
                _config.MinSelectorThreads, _config.MaxSelectorThreads);

            //so if we expect max of 10000 connections and we have 500 socketworkers, then each worker is expected
            //to handle 100 sockets. Load will be equally distributed as opposed to sequentially
            _workers = new DistributionPool(new SocketWorkerCreator(config.MaxConcurrentConns / config.MaxSocketWorkers, _selectors), 
                config.WaitForThread,  config.MaxSocketWorkers, config.MaxSocketWorkers);
        }

        /**
         * Attach the object passed (which is expected to be a SocketWorker) to one of the workers
         * in the DistributionPool. The object is attached to the worker with the least channels
         * attached to it.
         * If this is the first object being registed, then this also kicks off the workerselector
         * thread to check for data availability in each of the attached channels.
         * */
        internal void attach(object o)
        {
            PerfStat s = new PerfStat();
            s.start("SocketChannelSink:attach");
            Channel c = o as Channel;
            ChannelLogger.LOG.LogDebug("ServerChannelSink:attach:Attaching an object to the workers " + o + ":" + c.channelid);
            SocketWorker work = _workers.attach(o) as SocketWorker;
            s.checkpoint("SocketChannelSink:attach:attached worker");
            if (!work.myselector.running)
                work.myselector.myexecutor.execute(work.myselector.runTasks); //start to run only when first accept present
            s.dumpperf(ChannelLogger.LOG);
        }

        /**
         * Join to the threads present and wait for them to complete. This is just a helper function
         * and should be typically called only in the main thread if no other work is to be performed.
         * */
        public void join()
        {
            _svrExecutor.waitToComplete();
        }

        /**
         * This function is called when an event is streamed down a pipeline. This takes
         * different actions based on the event streamed down.
         * */
        public void sinkEvent(ChannelPipeline pipe, ChannelEvent e)
        {
            PerfStat s = new PerfStat();
            s.start("SocketChannelSink:sinkEvent");
            if (e is ChannelStateEvent)
            {
                ChannelLogger.LOG.LogDebug("ServerChannelSink:sinkEvent:Sinking ChannelStateEvent " + e);
                ChannelStateEvent evt = e as ChannelStateEvent;
                switch (evt.state)
                {
                    case ChannelStateEvent.ChannelState.BOUND:
                        //In this case we want to start off a thread to accept connections.
                        bind(evt.channel, (int)evt.value);
                        s.checkpoint("SocketChannelSink:sinkEvent:bound");
                        break;
                    default:
                        break;
                }
            }
            else if (e is MessageEvent)
            {
                //handle the write event.
                DownStreamMsgEvent msg = e as DownStreamMsgEvent;
                ByteBuffer obj = msg.getMessage() as ByteBuffer;
                AcceptedChannel channel = e.channel as AcceptedChannel;
                channel.queueData(obj);
                s.checkpoint("SocketChannelSink:sinkEvent:sent write message");
            }
            else
                ChannelLogger.LOG.LogDebug("ServerChannelSink:sinkEvent:Cannot sink event:" + e);
            s.dumpperf(ChannelLogger.LOG);
        }

        /**
         * This function does the real binding of the channel to the port. It fires off a 
         * channelbound event up, for other handlers to take action when a channel is bound.
         * This function kicks off the main worker to listen for incoming connections on
         * the channel.
         * */
        private void bind(Channel channel, int port)
        {
            //can only bind a ServerChannel
            ChannelLogger.LOG.LogDebug("ServerChannelSink:bind:Real binding of channel to " + port);
            if ((channel is ServerChannel) && (_mainWorker == null)) //the server is not started
            {
                ServerChannel svr = channel as ServerChannel;
                svr.server.bind(port);
                ChannelUtils.fireBoundEvent(channel);
                _mainWorker = new ServerWorker(svr, System.Threading.Timeout.Infinite); //for not don't timeout
                _svrExecutor.execute(_mainWorker.doWork); //start to run only when bound
            }
            else if (_mainWorker != null)
            {
                throw new Exception("Server Already Started");
            }
        }
    }

    /**
     * This class is a selector class to check for data availability for each of
     * the workers created to kick off a worker to do its work. So, when a ServerChannel
     * is bound, two threads are typically running, one to listen for connections and
     * one to check if data is available in the existing connections. If data is available
     * then upto the configured workers in the TuningParams threads are kicked off to 
     * execute the work for the sockets.
     * */
    class WorkerSelector : IDistributeObject
    {
        Executor _clientExecutor;       //The executor that will execute the various workers
        IList _workers;                 //The list of workers in which work has to be checked
        bool _complete;                 //flag to indicate if this selector should complete.
        internal TuningParams _config;           //configuration of the base channel based on which this selector shd work
        volatile bool _running;                  //If this selector is currently running. Define it as volatile to lock
        Executor _myExecutor;                   //The executor that will be used to schedule work for each socketworker

        /**
         * This can only be constructed by the SocketChannelSink. Will be constructed
         * to work as per the tuning parameters passed in. Will initiate the executor to 
         * be a threadpool executor which allow for preventing more than X threads to be
         * created.
         * */
        internal WorkerSelector(TuningParams config, Executor exec) : base(config.MaxWorkersPerSelector)
        {
            //String nm = typeof(SocketWorker).Name;
//            _clientExecutor = new ThreadPoolExecutor(nm, config.WaitForThread, config.NoSocketWorkers,
  //              config.NoSocketWorkers);
            _clientExecutor = exec;

            _workers = new ArrayList();
            _complete = false;
            _config = config;
            _running = false;
            _myExecutor = new AsynchronousExecutor(this.runTasks);
        }

        //access the executor for this selector
        internal Executor myexecutor { get { return _myExecutor; } }

        /**
         * Any worker that needs to be monitored by this selector has to register
         * using this function. This function is thread safe, the executing threadd
         * which accesses the registry will wait till this register is completed.
         * */
        internal void registerWorker(SocketWorker work)
        {
            lock (_workers)
            {
                _workers.Add(work);
            }
        }

        //the number of objects that this selector is handling
        public override int NoObjects { get { return _workers.Count;} }

        /**
         * When a sockerworker is attached to this workerselector 
         * register it with this object for monitoring.
         * */
        public override void attach(object o)
        {
            SocketWorker work = o as SocketWorker;
            ChannelLogger.LOG.LogDebug("WorkerSelector:attached:" + work.workerid);
            registerWorker(o as SocketWorker);
        }

        /**
         * The actual function that checks for availability of data in the workers.
         * This assumes that the workers provide a checkAvailable function which
         * indicates that data is available.
         * */
        public void runTasks()
        {
            _running = true;
            while (!_complete)
            {
                foreach (SocketWorker o in _workers)
                {
                    //distribute without checking, let the thread check, so we don't have to lock.
                    //if (o.checkAvailable(1000) > 0) //microsecs. TODO: configure
                    if (!o.running)
                    {
                        ChannelLogger.LOG.LogDebug("WorkerSelector:runTasks:The worker is not running, restarting:" + o.workerid);
                        _clientExecutor.execute(o.doWork);
                    }
                }

                //for now it is polling
                Thread.Sleep(_config.SelectorPollInterval);
            }
            _running = false;
        }

        /**
         * Attribute to check if this selector is running or not. Returns true if it is
         * running.
         * */
        public bool running { get { return _running; } }
    }

    /**
     * The creator object that will be used by the Distributed pool to create
     * the workerselector object when extending pool.
     * */
    public class WorkerSelectorCreator : IPoolableObjectCreator
    {
        TuningParams _config;           //The configuration for which this selector is created.
        Executor _exec;                 //The executor that will be used to execute the worker

        /**
         * Can be created only internally.
         * */
        internal WorkerSelectorCreator(TuningParams config, Executor exec)
        {
            _config = config;
            _exec = exec;
        }

        //Creates a new WorkerSelector
        public IPoolableObject create()
        {
            return new WorkerSelector(_config, _exec);
        }
    }

    /**
     * All registering of Sockets into the SocketWorker will be stoed as a task
     * and executed in synch with the monitoring of the sockets to prevent multiple
     * threads from accessing the selector object and hence preventing locking on the
     * selector object.
     * */
    internal class RegisterTask
    {
        AcceptedChannel _channel;       //The channel that has to be registered with the socketworker

        //can only be registered by the SocketWorker. This will create a dummy task.
        internal RegisterTask()
        {

        }

        //create a task to register the channel.
        internal RegisterTask(AcceptedChannel channel)
        {
            _channel = channel;
        }

        //access the channel to be registered
        internal AcceptedChannel registerchannel { get { return _channel; } }

        //check if it is dummy, if channel is null it is dummy
        internal bool dummy { get { return (_channel == null); } }
    }

    /**
     * This class provides the implementation of the work that has to be done when the
     * selector of the managed channels return that there is work to be done. Currently
     * this handles only the read and the write channels. This will fire the respective
     * events for the message to be handled once received or before send.
     * This can be a part of the distribution pool so that work is evenly distributed
     * across the workers. This class only handles the work from AcceptedChannel sockets.
     * */
    class SocketWorker : IDistributeObject
    {
        protected int _maxsocks;                    //max sockets this worker should handle
        protected SocketSelector _selector;         //The selector that will monitor the sockets
        volatile bool _isRunning = false;                    //if this worker is currently running
        protected WorkerSelector _mySelector;       //The selector that will monitor this worker and assign work
        private TSList<RegisterTask> _register;    //will hold the list of new channels to attach to
        private int _tasks;                         //the number of tasks of this worker
        ChannelPipeline _thisworkpipe;              //this will be used by all the channels handled by this worker
        int _workerid;                              //debugging purposes

        /**
         * Construction. This class is expected to be initialized only within this
         * namespace. It is initialized for a given max sockets to handle and the 
         * selector that will monitor and assign work.
         * */
        internal SocketWorker(int maxsocks, DistributionPool selector, int id) : base(maxsocks)
        {
            _maxsocks = maxsocks;
            _selector = new SocketSelector(_maxsocks, _workerid);
            _mySelector = selector.attach(this) as WorkerSelector;
            _register = new TSList<RegisterTask>(new RegisterTask());
            _tasks = 0;
            _workerid = id;
        }

        /**
         * This will be used by the distribution pool to attach a new channel to this
         * worker to monitor and handle. Currently the channel attached is only monitored
         * for READ. 
         * */
        public override void attach(object o)
        {
            ChannelLogger.LOG.LogDebug("SocketWorker:attach:Attaching the channel to worker." + o + ":" + _workerid);
            AcceptedChannel channel = o as AcceptedChannel;
            if (_thisworkpipe == null)
                _thisworkpipe = channel.pipeline; //the first time get the pipeline and use it.
            else
                channel.pipeline = _thisworkpipe; //else set it to what is here.
            _register.push(new RegisterTask(channel));
            _tasks++;
            //only read for now, when a message comes in, then add for write.
            //THis will register in a single thread in the doWork to prevent locking.
            //channel.clientsocket.register(_selector, SocketConsts.READ | SocketConsts.WRITE);
        }

        /**
         * Attribute to determine the number of objects that this worker is currently
         * monitoring.
         * */
        public override int NoObjects { get { return _tasks; } }

        //access to the selector for this object
        internal WorkerSelector myselector { get { return _mySelector; } }

        /**
         * Currently not Implemented.
         * */
        public void detach(AcceptedChannel channel)
        {
            //TODO: need to do correctly
        }

        /**
         * Waits for timeout period for some action available to be done on the sockets
         * registered with this worker. This function returns any new actions to be done
         * and any pending actions to be done.
         * */
        private int checkAvailable(int timeout)
        {
            PerfStat s = new PerfStat();
            s.start("SocketWorker:checkAvailable");
            int ret = 0;
            ret = _selector.select(timeout);
            s.dumpperf(ChannelLogger.LOG);
            return ret;
        }

        /**
         * Attribute to determine if this worker is currently running.
         * */
        public bool running { get { return _isRunning; } }

        /**
         * Actual function that reads from the sockets that have data to be read. This function
         * will fire the message received event to be handled by the different handlers
         * in the pipeline of the channel.
         * */
        public void doWork()
        {
            //have to lock here must be?
            //need to add timeout. have to configure this.
            PerfStat s = new PerfStat();
            s.start("SocketWorker:doWork");
            _isRunning = true;
            ChannelLogger.LOG.LogDebug("SocketWorker:doWork:Starting up worker thread for the different sockets:" + _workerid);
            //first register all the channels
            s.checkpoint("SocketWorker:doWork:registered");
            RegisterTask c = _register.pop();
            while (!c.dummy)
            {
                try
                {
                    ChannelLogger.LOG.LogDebug("SocketWorker:doWork:Registering socket:" + c.registerchannel.channelid + " with worker:" + _workerid);
                    c.registerchannel.clientsocket.register(_selector, SocketConsts.READ | SocketConsts.WRITE);
                    c = _register.pop();
                }
                catch (Exception e)
                {
                    //handle any exceptions in registration exception fired for each channel
                    ChannelLogger.LOG.LogError(e, e.Message);
                    ChannelUtils.fireExceptionEvent(c.registerchannel, e);
                }
            }

            s.checkpoint("SocketWorker:doWork:checkAvailable");
            if (checkAvailable(_mySelector._config.socketWaitForData) > 0)//do for one available.
            {
                s.checkpoint("SocketWorker:doWork:done");
                //once registered, this shd be fine
                //threading shd not be a problem
                IList selected = _selector.selectedKeys();
                foreach (object o in selected)
                {
                    SelectionKey key = o as SelectionKey;
                    ClientSocket sock = key.channel as ClientSocket;
                    AcceptedChannel channel = sock.attachment as AcceptedChannel;
                    try
                    {
                        //handle exception for each channel
                        SocketStream stream = new ChunkedStream(sock.socket);
                        if (key.readable && sock.stillConnected())
                        {
                            s.checkpoint("SocketWorker:doWork:readstream:" + channel.channelid + ":" + _workerid);
                            //This will only read the bytes that are available and will not bother about
                            //the completeness of the message. An appropriate Handler should be present
                            //in the chain of handlers through which this message is streamed that takes
                            //care of appending to the previous byte and appropriately handling.
                            //A standard handler will be provided for this which can be used.
                            ByteBuffer buffer = stream.read();
                            ChannelUtils.fireUpstreamMsgEvent(channel, buffer);
                            s.checkpoint("SocketWorker:doWork:read one Stream in :" + channel.channelid + ":" + _workerid);
                        }

                        //if (key.writable && sock.stillConnected())
                        //What if I did not wait for writable?
                        if (sock.writePresent && sock.stillConnected())
                        {
                            s.checkpoint("SocketWorker:doWork:startWrite:" + channel.channelid);
                            //write something here.
                            ChannelLogger.LOG.LogDebug("SocketWorker:doWork:Socket is ready to write:" + channel.channelid + ":" + _workerid);
                            //if more than one is queued, send it in the order it was received.
                            //pops in the reverse order, so put it into a list and send in the
                            //reverse order. No chance of threading problems here.
                            List<ByteBuffer> snd = new List<ByteBuffer>();
                            ByteBuffer write = channel.nextInQueue();
                            while (write != null)
                            {
                                //do not write directly here, since the order is reversed
                                //stream.write(write);
                                snd.Add(write);
                                write = channel.nextInQueue();
                            }

                            ChannelLogger.LOG.LogDebug("SocketWorker:doWork:Got :" + snd.Count + ": data objects to write:" + channel.channelid + ":" + _workerid);
                            for (int i = (snd.Count - 1); i >= 0; i--)
                            {
                                stream.write(snd[i]);
                                //fire off the sent event
                                ChannelUtils.fireWriteCompleted(channel, snd[i]);
                            }
                            s.checkpoint("SocketWorker:doWork:done one Write:" + channel.channelid);
                        }
                    }
                    catch (Exception e)
                    {
                        //handle any exceptions in reader
                        ChannelLogger.LOG.LogError(e, e.Message);
                        ChannelUtils.fireExceptionEvent(channel, e);
                    }

                    try
                    {
                        //irrespective of an exception, we want to go ahead and close the selec
                        //so the next continues and if not connected, disconnect.
                        _selector.doneSelect(key); //release key
                        s.checkpoint("SocketWorker:doWork:doneSelect:" + channel.channelid);
                        if (!sock.stillConnected())//possibly have to check later, else there might be a problem?
                        {
                            ChannelLogger.LOG.LogDebug("SocketWorker:doWork:Socket Not yet connected. Disconnecting." + sock.socket.LocalEndPoint.ToString() + ":" + channel.channelid);
                            //close the socket
                            //sock.close();
                            channel.close();
                        }
                    }
                    catch (Exception e)
                    {
                        //catch this as a separate exception
                        ChannelLogger.LOG.LogError(e, e.Message);
                        ChannelUtils.fireExceptionEvent(channel, e);
                    }
                }
            }
            ChannelLogger.LOG.LogDebug("SocketWorker:doWork:Done Reading available size." + _workerid);
            s.dumpperf(ChannelLogger.LOG);
            _isRunning = false;
        }

        public int workerid { get { return _workerid; } }
    }

    /**
     * A creator class for the socketworker to be used by the distribution pool to create
     * multiple workers to distribute the work to.
     * */
    class SocketWorkerCreator : IPoolableObjectCreator
    {
        int _maxsockets;                //The maximum sockets to be handled by each worker
        DistributionPool _selector;       //The selector that will handle the workers in this pool
        int _workers;                     //cnt of workers created

        /**
         * The constructor to set the context in which all the workers will be
         * created.
         * */
        public SocketWorkerCreator(int max, DistributionPool selector)
        {
            _maxsockets = max;
            _selector = selector;
            _workers = 0;
        }

        /**
         * Returns a new instance of the SocketWorker class based on the context
         * setup during the instantiation of this creator.
         * */
        public IPoolableObject create()
        {
            _workers++;
            return new SocketWorker(_maxsockets, _selector, _workers);
        }
    }

    /**
     * This class works on the server socket. listens to the port for connections and
     * handles any new connection coming into the server socket. This will hand off the
     * accepted socket to the workers of the distributed pool defined previously and will
     * do no other work other than accepting connections on the server channel.
     * */
    class ServerWorker
    {
        ServerChannel _server;              //The server on which to listen and accept conns
        int _timeout;                       //the timeout to wait for connection before continuing again to wait
        bool _complete = false;             //turn this flag on to stop the server.

        /**
         * This will be constructed within this namespace only and cannot be used outside.
         * This will instantiate this worker to work on channel and wait till timeout to
         * check for connections before checking if it has to complete. If not
         * it will continue waiting again.
         * */
        internal ServerWorker(ServerChannel channel, int timeout)
        {
            _server = channel;
            _timeout = timeout;
        }

        /**
         * Attribute to flag this worker to complete it's work and return.
         * */
        public bool complete { set { _complete = value; } }

        /**
         * The actual function that does the work of this class. This function calls
         * accept on the channel for a ClientSocket. If one received, it attaches it
         * to one of the workers of the sink that invoked this worker. It also
         * fires off a connection accepted event upstream for all the pipe handlers
         * to take action.
         * */
        public void doWork()
        {
            ChannelLogger.LOG.LogDebug("ServerWorker:doWork:Starting to do work");
            int channels = 0;
            while (!_complete)
            {
                try
                {
                    ClientSocket client = _server.server.accept(_timeout);
                    if (client != null)
                    {
                        //did not timeout
                        PerfStat s = new PerfStat();
                        s.start("ServerWorker:doWork");
                        ChannelLogger.LOG.LogDebug("ServerWorker:doWork:Got a new connection. Attaching to a worker");
                        ChannelPipeline pipe = _server.config.clientpipefactory.pipeline;
                        AcceptedChannel channel = new AcceptedChannel(client, _server.config as SocketChannelConfig,
                            pipe, channels);
                        channels++;
                        ServerChannelSink sink = _server.pipeline.sink as ServerChannelSink;
                        pipe.attach(channel, sink); //currently the sink is the same as the server sink, shd this be different?
                        sink.attach(channel);
                        s.dumpperf(ChannelLogger.LOG);
                        //TODO: fireoff an event here.
                        //worker.attach(channel);
                    }
                }
                catch (Exception e)
                {
                    //ignore timeout exception for now.
                    ChannelLogger.LOG.LogError(e, e.Message);
                    Console.WriteLine(e.Message);
                    Console.WriteLine(e.StackTrace);
                    ChannelUtils.fireExceptionEvent(_server, e);
                }
            }
        }
    }

}
